package com.bckj.fastboot.iot.mq.redis.delaymsg;

import com.bckj.fastboot.core.jackson.JsonUtils;
import com.bckj.fastboot.iot.common.dto.request.MessageRecordRequest;
import com.bckj.fastboot.iot.common.repository.MessageRecordRepository;
import com.bckj.fastboot.iot.mq.redis.constants.RedisMQConstants;
import com.bckj.fastboot.iot.mq.redis.model.DelayedMessage;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Set;

@Slf4j
@Component
public class DelayedMessageScheduler {

    @Resource
    private RedisTemplate<String, String> redisTemplate;
    @Resource
    private DelayedMessageHandler delayedMessageHandler;
    @Resource
    private MessageRecordRepository messageRecordRepository;

    /**
     * 每秒执行一次，扫描 ZSET 中 score <= 当前时间的消息
     */
    @Scheduled(fixedRate = 1000) // 每 1 秒执行一次
    public void processDelayedMessages() {
        long now = System.currentTimeMillis();

        // 查询所有 score <= now 的消息
        Set<String> messages = redisTemplate.opsForZSet().rangeByScore(RedisMQConstants.DELAYED_MESSAGES_ZSET_KEY, 0, now);

        if (messages != null && !messages.isEmpty()) {
            for (String messageJson : messages) {
                try {
                    log.info("🕒 执行延时消息: {}", messageJson);

                    // 处理消息（调用你的业务逻辑，比如关闭订单、发通知）
                    handleDelayedMessage(messageJson);
                } catch (Exception e) {
                    log.error("处理延时消息出错: {}", messageJson, e);
                    // 可以记录日志、重试等
                } finally {
                    // 处理成功后，从 ZSET 中移除该消息
                    redisTemplate.opsForZSet().remove(RedisMQConstants.DELAYED_MESSAGES_ZSET_KEY, messageJson);
                }
            }
        }
    }

    private void handleDelayedMessage(String messageJson) {
        log.info("[处理延时消息] 消息内容: {}", messageJson);
        DelayedMessage delayedMessage = JsonUtils.toBean(messageJson, DelayedMessage.class);
        try {
            delayedMessageHandler.handle(delayedMessage);
            updateMessageRecord(delayedMessage.getId(), 1);
        } catch (Exception e) {
            updateMessageRecord(delayedMessage.getId(), 2);
            throw new RuntimeException(e);
        }
    }

    @Async
    public void updateMessageRecord(String msgId, Integer status) {
        var request = new MessageRecordRequest();
        request.setMsgId(msgId);
        request.setStatus(status);
        messageRecordRepository.updateByMsgId(request);
    }
}